package cn.com.netedge.iot.frontend.comm.task;

import cn.com.netedge.iot.common.constants.MqConstants;
import cn.com.netedge.iot.common.utils.ChronicleQueueUtil;
import cn.com.netedge.iot.common.utils.StringUtil;
import cn.com.netedge.iot.frontend.comm.utils.JmsMsgUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Component("feCommLocalMessageConsumeTask")
public class LocalMessageConsumeTask {
    private static Logger logger = LoggerFactory.getLogger(LocalMessageConsumeTask.class);

    private static final int INTERVAL = 5 * 1000; // 5s
    private static final int BATCH_SIZE = 2000;

    private static List<String> queueNameList = new ArrayList<>(3);

    static {
        queueNameList.add(MqConstants.QUEUE_UP_MESSAGE_NORMAL);
        queueNameList.add(MqConstants.QUEUE_UP_MESSAGE_EMERGENCY);
    }

    @Scheduled(fixedDelay = INTERVAL)
    public void consume() {
        for (String queueName : queueNameList) {
            int count = 0;
            String message = ChronicleQueueUtil.dequeue(queueName);
            while (StringUtil.isNotEmpty(message) && count < BATCH_SIZE) {
                count++;
                int len = message.length();
                if (len > 256) {
                    len = 256;
                }
                logger.debug("consume local message -> queueName:{}, message:{} ", queueName, message.substring(0, len));
                if (!JmsMsgUtil.sendMessage(queueName, message)) {
                    break;
                }
                message = ChronicleQueueUtil.dequeue(queueName);
            }
        }
    }
}
